package local;

import data.handle.LineItemHandle;
import data.handle.PackMessage;
import data.handle.PartHandle;

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

public class HandlerThread implements Runnable {

    private LinkedBlockingQueue<PackMessage> lbq1;
    private LinkedBlockingQueue<PackMessage> lbq2;

    private PartHandle partHandle;
    private LineItemHandle lineItemHandle;

    private int fileReaderThreadNum;
    private int flag1 = 0;
    private int flag2 = 0;

    private AtomicLong counterFile1;
    private AtomicLong counterFile2;
    private CyclicBarrier cyclicBarrier;

    public HandlerThread(int fileReaderThreadNum, PartHandle partHandle, LineItemHandle lineItemHandle) {

        lbq1 = new LinkedBlockingQueue<>();
        lbq2 = new LinkedBlockingQueue<>();

        this.partHandle = partHandle;
        this.lineItemHandle = lineItemHandle;

        this.fileReaderThreadNum = fileReaderThreadNum;
    }



    public void setTest(AtomicLong counterFile1, AtomicLong counterFile2, CyclicBarrier cyclicBarrier){
        this.counterFile1 = counterFile1;
        this.counterFile2 = counterFile2;
        this.cyclicBarrier = cyclicBarrier;
    }

    @Override
    public void run() {
        try {
            while (true) {
                if (!lbq1.isEmpty()) {
                    PackMessage packMessage1 = lbq1.take();
                    if(packMessage1.isEndFlag()){
                        flag1++;
//                        System.out.println("end");
                    }
                    else {
//                        System.out.println("file1 " + "KEY: "+packMessage1.getPartKey()+"  brand: "+ packMessage1.getBrand());

                        partHandle.putBrandIndex(packMessage1.getPartKey(), (byte) packMessage1.getBrand());
                        counterFile1.incrementAndGet();
                    }
                }

                if(!lbq2.isEmpty()){
                    PackMessage packMessage2 = lbq2.take();
                    if(packMessage2.isEndFlag()){
                        flag2++;
//                        System.out.println("end");
                    }
                    else {

//                        System.out.println("file2 " + "KEY: "+packMessage2.getPartKey()+"  quantity: "+ packMessage2.getQuantity());
                        lineItemHandle.put(packMessage2.getPartKey(), packMessage2.getQuantity(), packMessage2.getPrice());
                        counterFile2.incrementAndGet();
                    }
                }

                if(flag2 == fileReaderThreadNum){
                    System.out.println("flag1 = " + flag1);
                    System.out.println("flag2 = " + flag2);
                    cyclicBarrier.await();
                    break;
                }


            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

//    public void addData(ByteBuffer line, int fileIndex) {
//        if (fileIndex == 0)
//            this.lbq1.add(line);
//        if (fileIndex == 1)
//            this.lbq2.add(line);
//    }

    public void addData(PackMessage packMessage, int fileIndex) {
        if (fileIndex == 0)
            this.lbq1.add(packMessage);
        if (fileIndex == 1)
            this.lbq2.add(packMessage);
    }
}
